package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.bean.ProductStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.util.ClickHouseUtil;
import com.atguigu.gmall.realtime.util.DateTimeUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @author: xu
 * @desc: 商品主题统计应用
 * 执行前需要启动的服务
 * -zk,kafka,logger.sh(nginx + 日志处理服务),maxwell,hdfs,hbase,Redis,ClichHouse
 * -BaseLogApp,BaseDBApp,OrderWideApp,PaymentWide,ProductStatsApp
 */
public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // 1.1  准备本地测试流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 设置并行度
        env.setParallelism(1);
        // 1.3 设置Checkpoint
        // env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // env.setStateBackend(new FsStateBackend("hdfs://node1:8020/gmall/checkpoint/ProductStatsApp"))
        // System.setProperty("HADOOP_USER_NAME", "root");

        // TODO 2.从Kafka中获取数据流
        // 2.1 声明相关的主题名称以及消费者组
        String groupId = "product_stats_app";
        String pageViewSourceTopic = "dwd_page_log";
        String favorInfoSourceTopic = "dwd_favor_info";
        String cartInfoSourceTopic = "dwd_cart_info";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSourceTopic = "dwm_payment_wide";
        String refundInfoSourceTopic = "dwd_order_refund_info";
        String commentInfoSourceTopic = "dwd_comment_info";
        // 2.2 从页面日志中获取点击和曝光数据
        DataStreamSource<String> pageViewStream = env.addSource(MyKafkaUtil.getKafkaSource(pageViewSourceTopic, groupId));
        // 2.3 从dwd_favor_info中获取收藏数据
        DataStreamSource<String> favorInfoStream = env.addSource(MyKafkaUtil.getKafkaSource(favorInfoSourceTopic, groupId));
        // 2.4 从dwd_cart_info中获取购物车数据
        DataStreamSource<String> cartInfoStream = env.addSource(MyKafkaUtil.getKafkaSource(cartInfoSourceTopic, groupId));
        // 2.5 从dwm_order_wide中获取订单数据
        DataStreamSource<String> orderWideStream = env.addSource(MyKafkaUtil.getKafkaSource(orderWideSourceTopic, groupId));
        // 2.6 从dwm_payment_wide中获取支付数据
        DataStreamSource<String> paymentWideStream = env.addSource(MyKafkaUtil.getKafkaSource(paymentWideSourceTopic, groupId));
        // 2.7 从dwd_order_refund_info中获取退款数据
        DataStreamSource<String> refundInfoStream = env.addSource(MyKafkaUtil.getKafkaSource(refundInfoSourceTopic, groupId));
        // 2.8 从dwd_order_refund_info中获取评价数据
        DataStreamSource<String> commentInfoStream = env.addSource(MyKafkaUtil.getKafkaSource(commentInfoSourceTopic, groupId));

        // TODO 3.将各个流的数据转换为统一的对象格式
        // 3.1 对点击和曝光数据进行转换  jsonStr-->ProduceStats
        SingleOutputStreamOperator<ProductStats> productClickAndDisplaysStream = pageViewStream.process(
                new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<ProductStats> out) throws Exception {
                        // 将json格式字符串转换为json对象
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                        String pageId = pageJsonObj.getString("page_id");
                        // 获取操作时间
                        Long ts = jsonObj.getLong("ts");
                        // 如果当前访问的页面是商品详情页，认为该商品被点击了一次
                        if ("good_detail".equals(pageId)) {
                            // 获取被点击商品的id，封装一次点击操作，然后向下游输出
                            Long skuId = pageJsonObj.getLong("item");
                            ProductStats productStats = ProductStats.builder().sku_id(skuId).click_ct(1L).ts(ts).build();
                            out.collect(productStats);
                        }
                        // 如果displays属性不为空，那么说明有曝光数据
                        JSONArray displays = jsonObj.getJSONArray("displays");
                        if (displays != null && displays.size() > 0) {
                            for (int i = 0; i < displays.size(); i++) {
                                // 获取曝光数据
                                JSONObject displayJsonObj = displays.getJSONObject(i);
                                // 判断是否曝光的某一个商品
                                if ("sku_id".equals(displayJsonObj.getString("item_type"))) {
                                    // 获取商品id，封装曝光商品对象，向下游输出
                                    Long skuId = displayJsonObj.getLong("item");
                                    ProductStats productStats = ProductStats.builder().sku_id(skuId).display_ct(1L).ts(ts).build();
                                    out.collect(productStats);
                                }
                            }
                        }
                    }
                }
        );

        // 3.2 转换收藏流数据
        SingleOutputStreamOperator<ProductStats> favorStatsStream = favorInfoStream.map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    return ProductStats.builder()
                            .sku_id(jsonObj.getLong("sku_id"))
                            .favor_ct(1L)
                            .ts(DateTimeUtil.stringToTimeStamp(jsonObj.getString("create_time")))
                            .build();
                }
        );

        // 3.3 转换购物车流数据
        SingleOutputStreamOperator<ProductStats> cartStatsStream = cartInfoStream.map(jsonStr -> {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    return ProductStats.builder()
                            .sku_id(jsonObj.getLong("sku_id"))
                            .cart_ct(1L)
                            .ts(DateTimeUtil.stringToTimeStamp(jsonObj.getString("create_time")))
                            .build();
                }
        );

        // 3.4 转换评价流数据
        SingleOutputStreamOperator<ProductStats> commonInfoStatsStream = commentInfoStream.map(jsonStr -> {
            JSONObject commonJsonObj = JSON.parseObject(jsonStr);
            String appraise = commonJsonObj.getString("appraise");
            Long goodCt = GmallConstant.APPRAISE_GOOD.equals(appraise) ? 1L : 0L;
            return ProductStats.builder()
                    .sku_id(commonJsonObj.getLong("sku_id"))
                    .comment_ct(1L)
                    .good_comment_ct(goodCt)
                    .ts(DateTimeUtil.stringToTimeStamp(commonJsonObj.getString("create_time")))
                    .build();
        });

        // 3.5 转换退款流数据
        SingleOutputStreamOperator<ProductStats> refundStatsStream = refundInfoStream.map(jsonStr -> {
            JSONObject refundJsonObj = JSON.parseObject(jsonStr);
            return ProductStats.builder()
                    .sku_id(refundJsonObj.getLong("sku_id"))
                    .refund_amount(refundJsonObj.getBigDecimal("refund_amount"))
                    .ts(DateTimeUtil.stringToTimeStamp(refundJsonObj.getString("create_time")))
                    .refundOrderIdSet(new HashSet(Collections.singleton(refundJsonObj.getLong("order_id"))))
                    .build();
        });

        // 3.6 转换支付流数据
        SingleOutputStreamOperator<ProductStats> paymentStatsStream = paymentWideStream.map(jsonStr -> {
                    PaymentWide paymentWide = JSON.parseObject(jsonStr, PaymentWide.class);
                    return ProductStats.builder()
                            .sku_id(paymentWide.getSku_id())
                            .payment_amount(paymentWide.getSplit_total_amount())
                            .ts(DateTimeUtil.stringToTimeStamp(paymentWide.getPayment_create_time()))
                            .paidOrderIdSet(new HashSet(Collections.singleton(paymentWide.getOrder_id())))
                            .build();
                }
        );

        // 3.7 对订单宽表进行转换
        SingleOutputStreamOperator<ProductStats> orderWideStatsStream = orderWideStream.map(jsonStr -> {
                    OrderWide orderWide = JSON.parseObject(jsonStr, OrderWide.class);
                    return ProductStats.builder()
                            .sku_id(orderWide.getSku_id())
                            .order_sku_num(orderWide.getSku_num())
                            .order_amount(orderWide.getSplit_total_amount())
                            .ts(DateTimeUtil.stringToTimeStamp(orderWide.getCreate_time()))
                            .orderIdSet(new HashSet(Collections.singleton(orderWide.getOrder_id())))
                            .build();
                }
        );

        // TODO 4. 将不同的流的数据通过union合并到一起
        DataStream<ProductStats> unionAllStream = productClickAndDisplaysStream.union(
                favorStatsStream, cartStatsStream, commonInfoStatsStream,
                refundStatsStream, paymentStatsStream, orderWideStatsStream
        );

        // TODO 5.设置Watermark，对数据进行分组、开窗、聚合
        SingleOutputStreamOperator<ProductStats> reduceStream = unionAllStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ProductStats>forMonotonousTimestamps()
                                .withTimestampAssigner((productStats, recordTimestamp) -> productStats.getTs())
                )
                // 注意：目前我们商品维度数据处理订单和支付宽表能够获取，其它的流式没有维度数据。所有我们这里使用skuId进行分组
                .keyBy(ProductStats::getSku_id)
                // 对分组之后的数据进行开窗 开一个10s的滚动窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 对窗口中的元素进行聚合
                .reduce(new ReduceFunction<ProductStats>() {
                            @Override
                            public ProductStats reduce(ProductStats stats1, ProductStats stats2) throws Exception {
                                stats1.setDisplay_ct(stats1.getDisplay_ct() + stats2.getDisplay_ct());
                                stats1.setClick_ct(stats1.getClick_ct() + stats2.getClick_ct());
                                stats1.setCart_ct(stats1.getCart_ct() + stats2.getCart_ct());
                                stats1.setFavor_ct(stats1.getFavor_ct() + stats2.getFavor_ct());
                                stats1.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));
                                stats1.getOrderIdSet().addAll(stats2.getOrderIdSet());
                                stats1.setOrder_ct((long) stats1.getOrderIdSet().size());
                                stats1.setOrder_sku_num(stats1.getOrder_sku_num() + stats2.getOrder_sku_num());
                                stats1.setPayment_amount(stats1.getPayment_amount().add(stats2.getPayment_amount()));

                                stats1.getRefundOrderIdSet().addAll(stats2.getRefundOrderIdSet());
                                stats1.setRefund_order_ct((long) stats1.getRefundOrderIdSet().size());
                                stats1.setRefund_amount(stats1.getRefund_amount().add(stats2.getRefund_amount()));

                                stats1.getPaidOrderIdSet().addAll(stats2.getPaidOrderIdSet());
                                stats1.setPaid_order_ct((long) stats1.getPaidOrderIdSet().size());

                                stats1.setComment_ct(stats1.getComment_ct() + stats2.getComment_ct());
                                stats1.setGood_comment_ct(stats1.getGood_comment_ct() + stats2.getGood_comment_ct());
                                return stats1;
                            }
                        },
                        new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                            @Override
                            public void process(Long key, Context context, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                                for (ProductStats productStats : elements) {
                                    productStats.setStt(DateTimeUtil.dateTimeToString(new Date(context.window().getStart())));
                                    productStats.setEdt(DateTimeUtil.dateTimeToString(new Date(context.window().getEnd())));
                                    productStats.setTs(System.currentTimeMillis());
                                    out.collect(productStats);
                                }
                            }
                        }
                );

        // TODO 6.补充商品的维度信息
        // 6.1 关联SKU维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSkuStream = AsyncDataStream.unorderedWait(
                reduceStream,
                new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getSku_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws Exception {
                        productStats.setSku_name(dimInfoJsonObj.getString("SKU_NAME"));
                        productStats.setSku_price(dimInfoJsonObj.getBigDecimal("PRICE"));
                        productStats.setSpu_id(dimInfoJsonObj.getLong("SPU_ID"));
                        productStats.setTm_id(dimInfoJsonObj.getLong("TM_ID"));
                        productStats.setCategory3_id(dimInfoJsonObj.getLong("CATEGORY3_ID"));
                    }
                },
                60, TimeUnit.SECONDS
        );
        // 6.2 关联SPU维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSpuStream = AsyncDataStream.unorderedWait(
                productStatsWithSkuStream,
                new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getSpu_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws Exception {
                        productStats.setSpu_name(dimInfoJsonObj.getString("SPU_NAME"));
                    }
                },
                60, TimeUnit.SECONDS
        );

        // 6.3 关联TM维度
        SingleOutputStreamOperator<ProductStats> productStatsWithTMStream = AsyncDataStream.unorderedWait(
                productStatsWithSpuStream,
                new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getTm_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws Exception {
                        productStats.setTm_name(dimInfoJsonObj.getString("TM_NAME"));
                    }
                },
                60, TimeUnit.SECONDS
        );

        // 6.4 关联Category维度
        SingleOutputStreamOperator<ProductStats> productStatsWithCategoryStream = AsyncDataStream.unorderedWait(
                productStatsWithTMStream,
                new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {
                    @Override
                    public String getKey(ProductStats productStats) {
                        return productStats.getCategory3_id().toString();
                    }

                    @Override
                    public void join(ProductStats productStats, JSONObject dimInfoJsonObj) throws Exception {
                        productStats.setCategory3_name(dimInfoJsonObj.getString("NAME"));
                    }
                },
                60, TimeUnit.SECONDS
        );
        productStatsWithCategoryStream.print(">>>>");

        // TODO 7.将聚合后的流数据写到ClickHouse中
        productStatsWithCategoryStream.addSink(
                ClickHouseUtil.<ProductStats>getJdbcSink("insert into product_stats values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        );

        env.execute(ProductStatsApp.class.getSimpleName());
    }
}

